// DemandPredictionReducer.java
package com.taxi.analysis.demand;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * 实时需求预测Reducer
 * 功能：基于分区数据进行需求预测分析
 * 技术：分区技术 + 需求预测算法
 */
public class DemandPredictionReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        String keyStr = key.toString();

        if (keyStr.startsWith("PARTITION_")) {
            processCompositePartition(keyStr, values, context);
        } else if (keyStr.startsWith("TIME_")) {
            processTimePartition(keyStr, values, context);
        } else if (keyStr.startsWith("GEO_")) {
            processGeoPartition(keyStr, values, context);
        } else if (keyStr.startsWith("DEMAND_")) {
            processDemandTypePartition(keyStr, values, context);
        }
    }

    /**
     * 处理综合分区数据
     */
    private void processCompositePartition(String partitionKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<CompositeDemand> demands = new ArrayList<>();

        // 解析综合需求数据
        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("COMPOSITE_DEMAND|")) {
                    String data = record.substring("COMPOSITE_DEMAND|".length());
                    CompositeDemand demand = parseCompositeDemand(data);
                    if (demand != null) {
                        demands.add(demand);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析综合需求错误: " + value.toString());
            }
        }

        if (!demands.isEmpty()) {
            // 综合需求预测分析
            CompositePrediction prediction = analyzeCompositeDemand(demands);

            // 输出综合预测结果
            outputCompositePrediction(partitionKey, prediction, context);
        }
    }

    /**
     * 处理时间分区数据
     */
    private void processTimePartition(String partitionKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<TimeDemand> demands = new ArrayList<>();

        // 解析时间需求数据
        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("TIME_DEMAND|")) {
                    String data = record.substring("TIME_DEMAND|".length());
                    TimeDemand demand = parseTimeDemand(data);
                    if (demand != null) {
                        demands.add(demand);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析时间需求错误: " + value.toString());
            }
        }

        if (!demands.isEmpty()) {
            // 时间需求预测分析
            TimePrediction prediction = analyzeTimeDemand(demands);

            // 输出时间预测结果
            outputTimePrediction(partitionKey, prediction, context);
        }
    }

    /**
     * 处理地理分区数据
     */
    private void processGeoPartition(String partitionKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<GeoDemand> demands = new ArrayList<>();

        // 解析地理需求数据
        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("GEO_DEMAND|")) {
                    String data = record.substring("GEO_DEMAND|".length());
                    GeoDemand demand = parseGeoDemand(data);
                    if (demand != null) {
                        demands.add(demand);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析地理需求错误: " + value.toString());
            }
        }

        if (!demands.isEmpty()) {
            // 地理需求预测分析
            GeoPrediction prediction = analyzeGeoDemand(demands);

            // 输出地理预测结果
            outputGeoPrediction(partitionKey, prediction, context);
        }
    }

    /**
     * 处理需求类型分区数据
     */
    private void processDemandTypePartition(String partitionKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<DemandType> demands = new ArrayList<>();

        // 解析需求类型数据
        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("DEMAND_TYPE|")) {
                    String data = record.substring("DEMAND_TYPE|".length());
                    DemandType demand = parseDemandType(data);
                    if (demand != null) {
                        demands.add(demand);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析需求类型错误: " + value.toString());
            }
        }

        if (!demands.isEmpty()) {
            // 需求类型预测分析
            DemandTypePrediction prediction = analyzeDemandType(demands);

            // 输出需求类型预测结果
            outputDemandTypePrediction(partitionKey, prediction, context);
        }
    }

    /**
     * 分析综合需求
     */
    private CompositePrediction analyzeCompositeDemand(List<CompositeDemand> demands) {
        CompositePrediction prediction = new CompositePrediction();

        Map<String, Integer> locationCounts = new HashMap<>();
        Map<String, Integer> timePeriodCounts = new HashMap<>();
        Map<String, Integer> dayTypeCounts = new HashMap<>();
        Map<String, Integer> densityCounts = new HashMap<>();
        Map<String, Integer> demandLevelCounts = new HashMap<>();
        Map<String, Integer> demandTrendCounts = new HashMap<>();

        double totalDistance = 0;
        double totalAmount = 0;
        double totalDuration = 0;
        double totalIntensity = 0;
        int totalPassengers = 0;

        for (CompositeDemand demand : demands) {
            locationCounts.put(demand.pickupLocationId, locationCounts.getOrDefault(demand.pickupLocationId, 0) + 1);
            timePeriodCounts.put(demand.timePeriod, timePeriodCounts.getOrDefault(demand.timePeriod, 0) + 1);
            dayTypeCounts.put(demand.dayType, dayTypeCounts.getOrDefault(demand.dayType, 0) + 1);
            densityCounts.put(demand.pickupDensity, densityCounts.getOrDefault(demand.pickupDensity, 0) + 1);
            demandLevelCounts.put(demand.demandLevel, demandLevelCounts.getOrDefault(demand.demandLevel, 0) + 1);
            demandTrendCounts.put(demand.demandTrend, demandTrendCounts.getOrDefault(demand.demandTrend, 0) + 1);

            totalDistance += demand.tripDistance;
            totalAmount += demand.totalAmount;
            totalDuration += demand.tripDuration;
            totalIntensity += demand.demandIntensity;
            totalPassengers += demand.passengerCount;
        }

        prediction.totalDemands = demands.size();
        prediction.avgDistance = totalDistance / demands.size();
        prediction.avgAmount = totalAmount / demands.size();
        prediction.avgDuration = totalDuration / demands.size();
        prediction.avgIntensity = totalIntensity / demands.size();
        prediction.avgPassengers = (double) totalPassengers / demands.size();

        prediction.topLocation = findMostFrequent(locationCounts);
        prediction.topTimePeriod = findMostFrequent(timePeriodCounts);
        prediction.topDayType = findMostFrequent(dayTypeCounts);
        prediction.topDensity = findMostFrequent(densityCounts);
        prediction.topDemandLevel = findMostFrequent(demandLevelCounts);
        prediction.topDemandTrend = findMostFrequent(demandTrendCounts);

        // 预测下一时段需求
        prediction.nextHourPrediction = predictNextHourDemand(demands);
        prediction.demandGrowthRate = calculateDemandGrowthRate(demands);
        prediction.riskLevel = assessDemandRisk(prediction);

        return prediction;
    }

    /**
     * 分析时间需求
     */
    private TimePrediction analyzeTimeDemand(List<TimeDemand> demands) {
        TimePrediction prediction = new TimePrediction();

        Map<String, Integer> locationCounts = new HashMap<>();
        Map<String, Integer> timePeriodCounts = new HashMap<>();
        Map<String, Integer> dayTypeCounts = new HashMap<>();
        Map<String, Integer> seasonCounts = new HashMap<>();
        Map<String, Integer> demandLevelCounts = new HashMap<>();

        double totalDistance = 0;
        double totalAmount = 0;
        double totalIntensity = 0;
        int totalHourlyDemand = 0;

        for (TimeDemand demand : demands) {
            locationCounts.put(demand.pickupLocationId, locationCounts.getOrDefault(demand.pickupLocationId, 0) + 1);
            timePeriodCounts.put(demand.timePeriod, timePeriodCounts.getOrDefault(demand.timePeriod, 0) + 1);
            dayTypeCounts.put(demand.dayType, dayTypeCounts.getOrDefault(demand.dayType, 0) + 1);
            seasonCounts.put(demand.season, seasonCounts.getOrDefault(demand.season, 0) + 1);
            demandLevelCounts.put(demand.demandLevel, demandLevelCounts.getOrDefault(demand.demandLevel, 0) + 1);

            totalDistance += demand.tripDistance;
            totalAmount += demand.totalAmount;
            totalIntensity += demand.demandIntensity;
            totalHourlyDemand += demand.hourlyDemandCount;
        }

        prediction.totalDemands = demands.size();
        prediction.avgDistance = totalDistance / demands.size();
        prediction.avgAmount = totalAmount / demands.size();
        prediction.avgIntensity = totalIntensity / demands.size();
        prediction.avgHourlyDemand = (double) totalHourlyDemand / demands.size();

        prediction.topLocation = findMostFrequent(locationCounts);
        prediction.topTimePeriod = findMostFrequent(timePeriodCounts);
        prediction.topDayType = findMostFrequent(dayTypeCounts);
        prediction.topSeason = findMostFrequent(seasonCounts);
        prediction.topDemandLevel = findMostFrequent(demandLevelCounts);

        // 时间模式预测
        prediction.peakHourPrediction = predictPeakHour(demands);
        prediction.timeEfficiency = calculateTimeEfficiency(prediction);

        return prediction;
    }

    /**
     * 分析地理需求
     */
    private GeoPrediction analyzeGeoDemand(List<GeoDemand> demands) {
        GeoPrediction prediction = new GeoPrediction();

        Map<String, Integer> pickupLocationCounts = new HashMap<>();
        Map<String, Integer> dropoffLocationCounts = new HashMap<>();
        Map<String, Integer> pickupDensityCounts = new HashMap<>();
        Map<String, Integer> dropoffDensityCounts = new HashMap<>();
        Map<String, Integer> timePeriodCounts = new HashMap<>();
        Map<String, Integer> demandLevelCounts = new HashMap<>();

        double totalDistance = 0;
        double totalAmount = 0;
        double totalIntensity = 0;

        for (GeoDemand demand : demands) {
            pickupLocationCounts.put(demand.pickupLocationId, pickupLocationCounts.getOrDefault(demand.pickupLocationId, 0) + 1);
            dropoffLocationCounts.put(demand.dropoffLocationId, dropoffLocationCounts.getOrDefault(demand.dropoffLocationId, 0) + 1);
            pickupDensityCounts.put(demand.pickupDensity, pickupDensityCounts.getOrDefault(demand.pickupDensity, 0) + 1);
            dropoffDensityCounts.put(demand.dropoffDensity, dropoffDensityCounts.getOrDefault(demand.dropoffDensity, 0) + 1);
            timePeriodCounts.put(demand.timePeriod, timePeriodCounts.getOrDefault(demand.timePeriod, 0) + 1);
            demandLevelCounts.put(demand.demandLevel, demandLevelCounts.getOrDefault(demand.demandLevel, 0) + 1);

            totalDistance += demand.tripDistance;
            totalAmount += demand.totalAmount;
            totalIntensity += demand.demandIntensity;
        }

        prediction.totalDemands = demands.size();
        prediction.avgDistance = totalDistance / demands.size();
        prediction.avgAmount = totalAmount / demands.size();
        prediction.avgIntensity = totalIntensity / demands.size();

        prediction.topPickupLocation = findMostFrequent(pickupLocationCounts);
        prediction.topDropoffLocation = findMostFrequent(dropoffLocationCounts);
        prediction.topPickupDensity = findMostFrequent(pickupDensityCounts);
        prediction.topDropoffDensity = findMostFrequent(dropoffDensityCounts);
        prediction.topTimePeriod = findMostFrequent(timePeriodCounts);
        prediction.topDemandLevel = findMostFrequent(demandLevelCounts);

        // 地理模式预测
        prediction.hotspotPrediction = predictHotspots(demands);
        prediction.spatialEfficiency = calculateSpatialEfficiency(prediction);

        return prediction;
    }

    /**
     * 分析需求类型
     */
    private DemandTypePrediction analyzeDemandType(List<DemandType> demands) {
        DemandTypePrediction prediction = new DemandTypePrediction();

        Map<String, Integer> fareLevelCounts = new HashMap<>();
        Map<String, Integer> distanceLevelCounts = new HashMap<>();
        Map<String, Integer> demandLevelCounts = new HashMap<>();
        Map<String, Integer> demandTrendCounts = new HashMap<>();
        Map<String, Integer> timePeriodCounts = new HashMap<>();

        double totalIntensity = 0;
        double totalAmount = 0;
        double totalDistance = 0;

        for (DemandType demand : demands) {
            fareLevelCounts.put(demand.fareLevel, fareLevelCounts.getOrDefault(demand.fareLevel, 0) + 1);
            distanceLevelCounts.put(demand.distanceLevel, distanceLevelCounts.getOrDefault(demand.distanceLevel, 0) + 1);
            demandLevelCounts.put(demand.demandLevel, demandLevelCounts.getOrDefault(demand.demandLevel, 0) + 1);
            demandTrendCounts.put(demand.demandTrend, demandTrendCounts.getOrDefault(demand.demandTrend, 0) + 1);
            timePeriodCounts.put(demand.timePeriod, timePeriodCounts.getOrDefault(demand.timePeriod, 0) + 1);

            totalIntensity += demand.demandIntensity;
            totalAmount += demand.totalAmount;
            totalDistance += demand.tripDistance;
        }

        prediction.totalDemands = demands.size();
        prediction.avgIntensity = totalIntensity / demands.size();
        prediction.avgAmount = totalAmount / demands.size();
        prediction.avgDistance = totalDistance / demands.size();

        prediction.topFareLevel = findMostFrequent(fareLevelCounts);
        prediction.topDistanceLevel = findMostFrequent(distanceLevelCounts);
        prediction.topDemandLevel = findMostFrequent(demandLevelCounts);
        prediction.topDemandTrend = findMostFrequent(demandTrendCounts);
        prediction.topTimePeriod = findMostFrequent(timePeriodCounts);

        // 需求类型预测
        prediction.trendPrediction = predictDemandTrend(demands);
        prediction.typeEfficiency = calculateTypeEfficiency(prediction);

        return prediction;
    }

    /**
     * 预测下一小时需求
     */
    private String predictNextHourDemand(List<CompositeDemand> demands) {
        Map<String, Integer> trendCounts = new HashMap<>();
        for (CompositeDemand demand : demands) {
            trendCounts.put(demand.demandTrend, trendCounts.getOrDefault(demand.demandTrend, 0) + 1);
        }

        String dominantTrend = findMostFrequentKey(trendCounts);
        double avgIntensity = demands.stream().mapToDouble(d -> d.demandIntensity).average().orElse(0);

        if ("上升".equals(dominantTrend) && avgIntensity > 50) {
            return "高需求增长";
        } else if ("上升".equals(dominantTrend)) {
            return "中等需求增长";
        } else if ("下降".equals(dominantTrend)) {
            return "需求下降";
        } else {
            return "需求稳定";
        }
    }

    /**
     * 计算需求增长率
     */
    private double calculateDemandGrowthRate(List<CompositeDemand> demands) {
        long upTrend = demands.stream().filter(d -> "上升".equals(d.demandTrend)).count();
        long downTrend = demands.stream().filter(d -> "下降".equals(d.demandTrend)).count();

        return ((double) upTrend - downTrend) / demands.size() * 100;
    }

    /**
     * 评估需求风险等级
     */
    private String assessDemandRisk(CompositePrediction prediction) {
        int riskScore = 0;

        if (prediction.avgIntensity > 80) riskScore += 3;
        else if (prediction.avgIntensity > 60) riskScore += 2;
        else if (prediction.avgIntensity > 40) riskScore += 1;

        if (prediction.demandGrowthRate > 50) riskScore += 3;
        else if (prediction.demandGrowthRate > 20) riskScore += 2;
        else if (prediction.demandGrowthRate > 0) riskScore += 1;

        if ("高需求增长".equals(prediction.nextHourPrediction)) riskScore += 2;
        else if ("中等需求增长".equals(prediction.nextHourPrediction)) riskScore += 1;

        if (riskScore >= 7) return "高风险";
        else if (riskScore >= 4) return "中风险";
        else if (riskScore >= 2) return "低风险";
        else return "正常";
    }

    /**
     * 预测高峰时段
     */
    private String predictPeakHour(List<TimeDemand> demands) {
        Map<String, Double> periodIntensity = new HashMap<>();
        Map<String, Integer> periodCounts = new HashMap<>();

        for (TimeDemand demand : demands) {
            periodIntensity.put(demand.timePeriod,
                    periodIntensity.getOrDefault(demand.timePeriod, 0.0) + demand.demandIntensity);
            periodCounts.put(demand.timePeriod,
                    periodCounts.getOrDefault(demand.timePeriod, 0) + 1);
        }

        String peakPeriod = "";
        double maxAvgIntensity = 0;

        for (Map.Entry<String, Double> entry : periodIntensity.entrySet()) {
            String period = entry.getKey();
            double avgIntensity = entry.getValue() / periodCounts.get(period);

            if (avgIntensity > maxAvgIntensity) {
                maxAvgIntensity = avgIntensity;
                peakPeriod = period;
            }
        }

        return peakPeriod + "(强度:" + String.format("%.1f", maxAvgIntensity) + ")";
    }

    /**
     * 计算时间效率
     */
    private double calculateTimeEfficiency(TimePrediction prediction) {
        // 基于平均需求强度和小时需求的效率计算
        return (prediction.avgIntensity / prediction.avgHourlyDemand) * 100;
    }

    /**
     * 预测热点区域
     */
    private String predictHotspots(List<GeoDemand> demands) {
        Map<String, Double> locationIntensity = new HashMap<>();
        Map<String, Integer> locationCounts = new HashMap<>();

        for (GeoDemand demand : demands) {
            locationIntensity.put(demand.pickupLocationId,
                    locationIntensity.getOrDefault(demand.pickupLocationId, 0.0) + demand.demandIntensity);
            locationCounts.put(demand.pickupLocationId,
                    locationCounts.getOrDefault(demand.pickupLocationId, 0) + 1);
        }

        List<Map.Entry<String, Double>> sortedLocations = new ArrayList<>();
        for (Map.Entry<String, Double> entry : locationIntensity.entrySet()) {
            String location = entry.getKey();
            double avgIntensity = entry.getValue() / locationCounts.get(location);
            sortedLocations.add(new AbstractMap.SimpleEntry<>(location, avgIntensity));
        }

        sortedLocations.sort((a, b) -> Double.compare(b.getValue(), a.getValue()));

        StringBuilder hotspots = new StringBuilder();
        for (int i = 0; i < Math.min(3, sortedLocations.size()); i++) {
            if (i > 0) hotspots.append(",");
            hotspots.append(sortedLocations.get(i).getKey())
                    .append("(").append(String.format("%.1f", sortedLocations.get(i).getValue())).append(")");
        }

        return hotspots.toString();
    }

    /**
     * 计算空间效率
     */
    private double calculateSpatialEfficiency(GeoPrediction prediction) {
        // 基于平均距离和需求强度的空间效率计算
        return (prediction.avgIntensity / prediction.avgDistance) * 10;
    }

    /**
     * 预测需求趋势
     */
    private String predictDemandTrend(List<DemandType> demands) {
        Map<String, Integer> trendCounts = new HashMap<>();
        for (DemandType demand : demands) {
            trendCounts.put(demand.demandTrend, trendCounts.getOrDefault(demand.demandTrend, 0) + 1);
        }

        String dominantTrend = findMostFrequentKey(trendCounts);
        int dominantCount = trendCounts.get(dominantTrend);
        double dominantRatio = (double) dominantCount / demands.size();

        return dominantTrend + "(占比:" + String.format("%.1f", dominantRatio * 100) + "%)";
    }

    /**
     * 计算类型效率
     */
    private double calculateTypeEfficiency(DemandTypePrediction prediction) {
        // 基于需求强度和平均费用的类型效率计算
        return (prediction.avgIntensity / prediction.avgAmount) * 100;
    }

    /**
     * 输出综合预测结果
     */
    private void outputCompositePrediction(String partitionKey, CompositePrediction prediction, Context context)
            throws IOException, InterruptedException {

        String predictionInfo = String.format("综合预测|总需求:%d|平均距离:%.2f|平均费用:%.2f|平均时长:%.2f|平均强度:%.2f|平均乘客:%.1f|热门地点:%s|热门时段:%s|热门日期:%s|热门密度:%s|主要需求:%s|主要趋势:%s|下时预测:%s|增长率:%.1f%%|风险等级:%s",
                prediction.totalDemands, prediction.avgDistance, prediction.avgAmount,
                prediction.avgDuration, prediction.avgIntensity, prediction.avgPassengers,
                prediction.topLocation, prediction.topTimePeriod, prediction.topDayType,
                prediction.topDensity, prediction.topDemandLevel, prediction.topDemandTrend,
                prediction.nextHourPrediction, prediction.demandGrowthRate, prediction.riskLevel);

        context.write(new Text("综合预测_" + partitionKey), new Text(predictionInfo));
    }

    /**
     * 输出时间预测结果
     */
    private void outputTimePrediction(String partitionKey, TimePrediction prediction, Context context)
            throws IOException, InterruptedException {

        String predictionInfo = String.format("时间预测|总需求:%d|平均距离:%.2f|平均费用:%.2f|平均强度:%.2f|平均小时需求:%.1f|热门地点:%s|热门时段:%s|热门日期:%s|热门季节:%s|主要需求:%s|高峰预测:%s|时间效率:%.2f",
                prediction.totalDemands, prediction.avgDistance, prediction.avgAmount,
                prediction.avgIntensity, prediction.avgHourlyDemand, prediction.topLocation,
                prediction.topTimePeriod, prediction.topDayType, prediction.topSeason,
                prediction.topDemandLevel, prediction.peakHourPrediction, prediction.timeEfficiency);

        context.write(new Text("时间预测_" + partitionKey), new Text(predictionInfo));
    }

    /**
     * 输出地理预测结果
     */
    private void outputGeoPrediction(String partitionKey, GeoPrediction prediction, Context context)
            throws IOException, InterruptedException {

        String predictionInfo = String.format("地理预测|总需求:%d|平均距离:%.2f|平均费用:%.2f|平均强度:%.2f|热门起点:%s|热门终点:%s|起点密度:%s|终点密度:%s|热门时段:%s|主要需求:%s|热点预测:%s|空间效率:%.2f",
                prediction.totalDemands, prediction.avgDistance, prediction.avgAmount,
                prediction.avgIntensity, prediction.topPickupLocation, prediction.topDropoffLocation,
                prediction.topPickupDensity, prediction.topDropoffDensity, prediction.topTimePeriod,
                prediction.topDemandLevel, prediction.hotspotPrediction, prediction.spatialEfficiency);

        context.write(new Text("地理预测_" + partitionKey), new Text(predictionInfo));
    }

    /**
     * 输出需求类型预测结果
     */
    private void outputDemandTypePrediction(String partitionKey, DemandTypePrediction prediction, Context context)
            throws IOException, InterruptedException {

        String predictionInfo = String.format("类型预测|总需求:%d|平均强度:%.2f|平均费用:%.2f|平均距离:%.2f|主要费用:%s|主要距离:%s|主要需求:%s|主要趋势:%s|热门时段:%s|趋势预测:%s|类型效率:%.2f",
                prediction.totalDemands, prediction.avgIntensity, prediction.avgAmount,
                prediction.avgDistance, prediction.topFareLevel, prediction.topDistanceLevel,
                prediction.topDemandLevel, prediction.topDemandTrend, prediction.topTimePeriod,
                prediction.trendPrediction, prediction.typeEfficiency);

        context.write(new Text("类型预测_" + partitionKey), new Text(predictionInfo));
    }

    /**
     * 找出最频繁的项目
     */
    private String findMostFrequent(Map<String, Integer> counts) {
        return counts.entrySet().stream()
                .max(Map.Entry.comparingByValue())
                .map(entry -> entry.getKey() + "(" + entry.getValue() + "次)")
                .orElse("未知");
    }

    /**
     * 找出最频繁的键
     */
    private String findMostFrequentKey(Map<String, Integer> counts) {
        return counts.entrySet().stream()
                .max(Map.Entry.comparingByValue())
                .map(Map.Entry::getKey)
                .orElse("未知");
    }

    // 解析方法和内部类定义
    private CompositeDemand parseCompositeDemand(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 17) {
                CompositeDemand demand = new CompositeDemand();
                demand.pickupLocationId = fields[0];
                demand.dropoffLocationId = fields[1];
                demand.pickupDatetime = fields[2];
                demand.pickupHour = Integer.parseInt(fields[3]);
                demand.pickupMinute = Integer.parseInt(fields[4]);
                demand.timePeriod = fields[5];
                demand.dayType = fields[6];
                demand.tripDistance = Double.parseDouble(fields[7]);
                demand.totalAmount = Double.parseDouble(fields[8]);
                demand.tripDuration = Double.parseDouble(fields[9]);
                demand.passengerCount = Integer.parseInt(fields[10]);
                demand.pickupDensity = fields[11];
                demand.fareLevel = fields[12];
                demand.distanceLevel = fields[13];
                demand.demandIntensity = Double.parseDouble(fields[14]);
                demand.demandLevel = fields[15];
                demand.demandTrend = fields[16];
                return demand;
            }
        } catch (Exception e) {
            System.err.println("解析综合需求错误: " + data);
        }
        return null;
    }

    private TimeDemand parseTimeDemand(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 10) {
                TimeDemand demand = new TimeDemand();
                demand.pickupLocationId = fields[0];
                demand.pickupDatetime = fields[1];
                demand.timePeriod = fields[2];
                demand.dayType = fields[3];
                demand.season = fields[4];
                demand.tripDistance = Double.parseDouble(fields[5]);
                demand.totalAmount = Double.parseDouble(fields[6]);
                demand.demandIntensity = Double.parseDouble(fields[7]);
                demand.demandLevel = fields[8];
                demand.hourlyDemandCount = Integer.parseInt(fields[9]);
                return demand;
            }
        } catch (Exception e) {
            System.err.println("解析时间需求错误: " + data);
        }
        return null;
    }

    private GeoDemand parseGeoDemand(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 11) {
                GeoDemand demand = new GeoDemand();
                demand.pickupLocationId = fields[0];
                demand.dropoffLocationId = fields[1];
                demand.pickupHour = Integer.parseInt(fields[2]);
                demand.timePeriod = fields[3];
                demand.dayType = fields[4];
                demand.tripDistance = Double.parseDouble(fields[5]);
                demand.totalAmount = Double.parseDouble(fields[6]);
                demand.pickupDensity = fields[7];
                demand.dropoffDensity = fields[8];
                demand.demandIntensity = Double.parseDouble(fields[9]);
                demand.demandLevel = fields[10];
                return demand;
            }
        } catch (Exception e) {
            System.err.println("解析地理需求错误: " + data);
        }
        return null;
    }

    private DemandType parseDemandType(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 10) {
                DemandType demand = new DemandType();
                demand.pickupLocationId = fields[0];
                demand.pickupHour = Integer.parseInt(fields[1]);
                demand.timePeriod = fields[2];
                demand.fareLevel = fields[3];
                demand.distanceLevel = fields[4];
                demand.demandLevel = fields[5];
                demand.demandTrend = fields[6];
                demand.demandIntensity = Double.parseDouble(fields[7]);
                demand.totalAmount = Double.parseDouble(fields[8]);
                demand.tripDistance = Double.parseDouble(fields[9]);
                return demand;
            }
        } catch (Exception e) {
            System.err.println("解析需求类型错误: " + data);
        }
        return null;
    }

    // 内部类定义
    private static class CompositeDemand {
        String pickupLocationId, dropoffLocationId, pickupDatetime, timePeriod, dayType;
        String pickupDensity, fareLevel, distanceLevel, demandLevel, demandTrend;
        int pickupHour, pickupMinute, passengerCount;
        double tripDistance, totalAmount, tripDuration, demandIntensity;
    }

    private static class TimeDemand {
        String pickupLocationId, pickupDatetime, timePeriod, dayType, season, demandLevel;
        double tripDistance, totalAmount, demandIntensity;
        int hourlyDemandCount;
    }

    private static class GeoDemand {
        String pickupLocationId, dropoffLocationId, timePeriod, dayType;
        String pickupDensity, dropoffDensity, demandLevel;
        int pickupHour;
        double tripDistance, totalAmount, demandIntensity;
    }

    private static class DemandType {
        String pickupLocationId, timePeriod, fareLevel, distanceLevel, demandLevel, demandTrend;
        int pickupHour;
        double demandIntensity, totalAmount, tripDistance;
    }

    private static class CompositePrediction {
        int totalDemands;
        double avgDistance, avgAmount, avgDuration, avgIntensity, avgPassengers;
        String topLocation, topTimePeriod, topDayType, topDensity, topDemandLevel, topDemandTrend;
        String nextHourPrediction, riskLevel;
        double demandGrowthRate;
    }

    private static class TimePrediction {
        int totalDemands;
        double avgDistance, avgAmount, avgIntensity, avgHourlyDemand, timeEfficiency;
        String topLocation, topTimePeriod, topDayType, topSeason, topDemandLevel, peakHourPrediction;
    }

    private static class GeoPrediction {
        int totalDemands;
        double avgDistance, avgAmount, avgIntensity, spatialEfficiency;
        String topPickupLocation, topDropoffLocation, topPickupDensity, topDropoffDensity;
        String topTimePeriod, topDemandLevel, hotspotPrediction;
    }

    private static class DemandTypePrediction {
        int totalDemands;
        double avgIntensity, avgAmount, avgDistance, typeEfficiency;
        String topFareLevel, topDistanceLevel, topDemandLevel, topDemandTrend;
        String topTimePeriod, trendPrediction;
    }
}
